package DianShang_2024.ds_01.tzgc

import groovy.sql.DataSet
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.feature.{OneHotEncoder, StandardScaler, VectorAssembler}
import org.apache.spark.ml.linalg.{DenseVector, SparseVector}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.{Window, WindowSpec}
import org.apache.spark.sql.functions.{asc, dense_rank, expr}

import java.util.Properties

object tzgc02 {
  def main(args: Array[String]): Unit = {
    /*
          2、根据Hive的dwd库中相关表或MySQL中shtd_store中相关商品表（sku_info），获取id、spu_id、price、weight、tm_id、category3_id 这六个
          字段并进行数据预处理，对price、weight进行规范化(StandardScaler)处理，对spu_id、tm_id、category3_id进行one-hot编码处
          理（若该商品属于该品牌则置为1，否则置为0）,并按照id进行升序排序，在集群中输出第一条数据前10列（无需展示字段名），将结果截图粘贴至
          客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
    字段	类型	中文含义	备注
    id	double	主键
    price	double	价格
    weight	double	重量
    spu_id#1	double	spu_id 1	若属于该spu_id，则内容为1否则为0
    spu_id#2	double	spu_id 2	若属于该spu_id，则内容为1否则为0
    .....	double
    tm_id#1	double	品牌1	若属于该品牌，则内容为1否则为0
    tm_id#2	double	品牌2	若属于该品牌，则内容为1否则为0
    ……	double
    category3_id#1	double	分类级别3 1	若属于该分类级别3，则内容为1否则为0
    category3_id#2	double	分类级别3 2	若属于该分类级别3，则内容为1否则为0
    ……

    结果格式如下：
    --------------------第一条数据前10列结果展示为：---------------------
    1.0,0.892346,1.72568,0.0,0.0,0.0,0.0,1.0,0.0,0.0
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第一套卷子特征工程第二题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  定义窗口规范,即窗口里面的数据按照当前字段进行排序或者聚合分析等等
    val window01:WindowSpec=Window.orderBy("spu_id")
    val window02:WindowSpec=Window.orderBy("tm_id")
    val window03:WindowSpec=Window.orderBy("category3_id")

    //  拿到起始数据
    /*
        1.dense_rank() 是一个窗口函数，用于为每个分组的行分配一个排名值，排名相同的行将获得相同的排名，且不会跳过排名。
    2.over(window01) 指定了窗口函数的计算范围（这里假设 window01 是之前定义的窗口规范）。
    最后，- 1 是将排名值减去 1，可能是为了从 0 开始索引。
    3.dense_rank().over(window03)然后这两个合起来就是为每一行分配一个唯一的索引，起到的效果和StringIndexer是一样的，都是为了后面的one-hot编码
    做准备
    4.setDropLast(true): 在生成独热编码时，最后一个类别的列将被丢弃。这在许多机器学习模型中是常见的做法，因为最后一个类别的存在可以通过其他列的值推断出来，从而减少冗余信息和模型复杂性。

    setDropLast(false): 所有类别都会被保留，包括最后一个类别的列。这可能会导致特征维度增加，但在某些情况下，这样做可能更合适，特别是在某些需要完整类别信息的应用场景中。
     */
    val source=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","sku_info",connect)
      .select("id","spu_id","price","weight","tm_id","category3_id")
      .withColumn("spu_id_index",dense_rank().over(window01) -1)
      .withColumn("tm_id_index",dense_rank().over(window02) -1)
      .withColumn("category3_id_index",dense_rank().over(window03) -1)

    //  将表格写入hive，为了后面的推荐系统使用
    source.write.saveAsTable("xiaoqiu.sku_info_vector")
    source.show

    //  对spu_id、tm_id、category3_id进行one-hot编码处理（若该商品属于该品牌则置为1，否则置为0）
    val hotEncoder:OneHotEncoder=new OneHotEncoder()
      .setInputCols(Array("spu_id_index","tm_id_index","category3_id_index"))
      .setOutputCols(Array("spu_id_hot","tm_id_hot","category3_id_hot"))
      .setDropLast(false)

    //  规范化处理首先要对字段列组合为单一的特征向量(vectorAssmbler),然后进行标准化处理(standardScaler)

    //  转化为特征向量 vector:向量       Assembler:汇编程序
    val vectorAssembler1:VectorAssembler=new VectorAssembler()
      .setInputCols(Array("price"))
      .setOutputCol("price_v")

    val vectorAssembler2:VectorAssembler=new VectorAssembler()
      .setInputCols(Array("weight"))
      .setOutputCol("weight_v")

    //  标准化处理器(standardScaler) standard:标准   scaler:定标器
    val standardScaler1:StandardScaler=new StandardScaler()
      .setInputCol("price_v")
      .setOutputCol("price_sca")
      .setWithMean(true)

    val standardScaler2:StandardScaler=new StandardScaler()
      .setInputCol("weight_v")
      .setOutputCol("weight_sca")
      .setWithMean(true)

    //  定义类似一个管道，定义数据处理的流程pipeline:管道
    //  fit(source)：这个方法会训练Pipeline。source 是输入数据集，通常是一个DataFrame。Pipeline会依次应用每个阶段的转换，并最终生成一个 PipelineModel，这个模型包含了所有经过训练的阶段的具体实现。
    val pipelineModel:PipelineModel=new Pipeline()
      .setStages(Array(vectorAssembler1,vectorAssembler2,standardScaler1,standardScaler2,hotEncoder))
      .fit(source)



    //  spark的udf即用户自定义方法
    //  SparseVector:稀疏向量,下面的方法就是将稀疏向量转化成数组，然后转化为字符串以逗号隔开sparse:稀疏的
    spark.udf.register(
      "vectorToArray",
      (v1:SparseVector) =>{ v1.toArray.mkString(",")}
    )

    //  下面的方法是取到稠密向量的第一个元素
    //  v1.apply(0) 通过索引 0 来获取稠密向量的第一个元素。     dense:密集的
    spark.udf.register(
      "vectorToDouble",
      (v1:DenseVector) => {  v1.apply(0)  }
    )


    /*
        expr 的作用
    表达式解析：expr 函数允许你使用 SQL 语法来定义列的计算方式。它能够将字符串形式的表达式解析为一个可执行的 Spark SQL 表达式。
    增强可读性：使用 SQL 语法使得表达式更具可读性，特别是对于复杂的计算或操作。
    示例中的具体用法

    one-hot编码处理后的数据格式为:(12,[3],[1.0]),这也是稀疏向量的一种写法，表示长度为12，索引为3位置的值为1
     */
    val result=pipelineModel.transform(source)
      .withColumn("spu_id_hot",expr("vectorToArray(spu_id_hot)"))
      .withColumn("tm_id_hot",expr("vectorToArray(tm_id_hot)"))
      .withColumn("category3_id_hot",expr("vectorToArray(category3_id_hot)"))
      .withColumn("price_sca",expr("vectorToDouble(price_sca)"))
      .withColumn("weight_sca",expr("vectorToDouble(weight_sca)"))
      .select("id","price_sca","weight_sca","spu_id_hot","tm_id_hot","category3_id_hot")
      .orderBy(asc("id"))
      .limit(1)



    println("初步结果")
    result.show(false)




    println("------------------第一条数据前10列结果为:-----------------------------")


    //  写法1
//    result.collect.foreach(
//      r=> {
//        println(r.toSeq.flatMap(
//          r => r.toString.split(",")
//        ).take(10).mkString(","))
//      })

//  写法2
    val r1: Array[Row] = result.collect()
    r1.foreach {
      r => println(r.mkString(",").split(",").take(10).mkString(","))
    }


    spark.close()



  }

}
